package com.atguigu.realtime.function;

import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import com.atguigu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/14 14:35
 * 通用的维度表异步处理的函数
 * <p>
 * T表示维度处理的时候, 流中的数据类型
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor pool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        pool = ThreadPoolUtil.getThreadPool();
    }
    
    protected abstract void addDim(Jedis redisClient, Connection phoenixConn, T input, ResultFuture<T> resultFuture) throws Exception;
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        // 多线程+多客户端 来解决异步没有异步请求客户端
        pool.submit(new Runnable() {
            
            @Override
            public void run() {
                
                Connection phoenixConn = null;
                Jedis redisClient = null;
                // 所有读取维度的操作
                try {
                    // 1. 创建Phoenix客户端和redis客户端
                    phoenixConn = JdbcUtil.getConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
                    redisClient = RedisUtil.getRedisClient();
                    
                    // 2. 读取维度数据
                    // 不知道读取那些维度表, 也没有办法获取到对应的id...
                    addDim(redisClient, phoenixConn, input, resultFuture);  // 书写具体的业务
                    
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (phoenixConn != null) {
                        try {
                            phoenixConn.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                    
                    if (redisClient != null) {
                        redisClient.close();
                    }
                }
                
            }
        });
    }
    
}
